package com.hliushi.spark.sql

import org.apache.spark.sql.types._
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.junit.Test

/**
 * descriptions:
 *
 * author: Hliushi
 * date: 2021/5/19 16:11
 */
class NullProcessor {
  /**
   * 常见的RelationalGroupedDataset 获取方式有三种
   * .    groupBy
   * .    rollup
   * .    cube
   * 无论通过任何一种方式获取了RelationGroupedDataset对象, 其所表示是的都是是一个被分组的DataFrame
   * 通过这个对象, 可以对数据集的分组结果进行聚合
   * #
   * 需要注意的是, RelationGroupedDataset并不是DataFrame, 所以其中并没有DataFrame的方法,
   * 只有如下一些聚合相关的方法, 这些方法调用之后会生成DataFrame对象, 然后就可以使用DataFrame的算子进行操作了
   * . avg    count    max   min     mean    sum
   * . agg  聚合, 可以使用sql.functions中的函数俩配合进行操作
   * .      pmDF.groupBy($"year")
   * .          .agg(avg($"pm") as "pm_avg")
   */

  // 1.创建SparkSession
  val spark: SparkSession = SparkSession.builder()
    .appName("null_processor")
    .master("local[6]")
    .getOrCreate()

  import org.apache.spark.sql.functions._
  import spark.implicits._

  val source: DataFrame = spark.read
    .option("header", value = true)
    .option("delimiter", value = ",")
    .option("inferSchema", value = true)
    .csv("dataset/house_prices_train.csv")

  /**
   * 空值处理
   * .    去掉空值列
   * .    填充其他值
   * .    扩展
   */
  @Test
  def identifyNullNaN(): Unit = {
    spark.sparkContext.setLogLevel("warn")
    source.createOrReplaceTempView("house_price")

    /**
     * 识别空值, 查看不重复的取值是多少个, 分别是什么
     * countDistinct 聚合某一列的不重复的值数量
     * 通过for循环多次打印, 每一列的取值范围
     * 针对每一列查看取值范围, 找到空值类型
     * 所有列中空值都是字符串 "NA"
     */
    for (c <- source.columns) {
      source.groupBy(col(c))
        .agg(first(col(c)) as c, countDistinct(col(c)) as s"${c}_count")
        .show()
      // 有缺失值的列为 Alley(小路), MasVnrType(表层砌体), BsmtQual(地下室高度)
    }
  }

  /**
   * 处理空值
   * .  1.找到空值含量
   * .  2.如果比较大, 则删除整列
   * .  3.如果不算太大, 填充值
   */
  @Test
  def countNA(): Unit = {
    /**
     * 通过 'Alley === "NA" 取得一个Boolean列, 把Boolean转为Int, true -> 1, false -> 0
     * sum的话, 就是把所有的1 加了起来, 直接求得列中 true 的数据, 也就是 NA 的数量
     */
    source.agg(sum(($"Alley" === "NA").cast(IntegerType)) / count($"Alley") as "NA_Alley")
      .show()

    //  Alley列的空值比例为 93%
    //  +------------------+
    //  |          NA_Alley|
    //  +------------------+
    //  |0.9376712328767123|
    //  +------------------+
    source.agg(sum(($"BsmtQual" === "NA").cast(IntegerType)) / count($"MasVnrType") as "NA_BsmtQual")
      .show()
    //  BsmtQual列的空值比例为 2.5%
    //  +--------------------+
    //  |         NA_BsmtQual|
    //  +--------------------+
    //  |0.025342465753424658|
    //  +--------------------+
  }


  /**
   * 空值类型分为三种
   * .      null    NaN     "NA"
   * 过滤缺失值的方式
   * .    1.如果一列数据有太多缺失值, 删除此列
   * .    2.使用均值, 中值, 众数, 随机数等代替缺失值, 但是不推荐,
   * .          因为这就等于认为添加噪声, 降低数据质量
   * .    3.拆值, 比如说性别有男, 女, 缺失, 三种状态, 拆成三列,
   * .          是否男, 是否女, 是否缺失, 这样最为精确
   */
  @Test
  def dropNaAndFillNa(): Unit = {
    spark.sparkContext.setLogLevel("warn")
    val innerSource = spark.read
      .option("header", value = true)
      .option("inferSchema", value = true)
      .csv("dataset/null_file_test.csv")

    innerSource.show()
    //  在没有删除任何行的时候出现这种情况
    //  +-------+---------+---+----+---------+
    //  |   name|    alias|age| pga|otherInfo|
    //  +-------+---------+---+----+---------+
    //  |ulysses| thompson| 64| 1.9|     null|
    //  |  katie|   carson| 25|3.65|     null|
    //  |   luke|     king| 65|0.73|     null|
    //  |  holly| davidson| 57|2.43|    hello|
    //  |   fred|   miller| 55|3.77|     null|
    //  |  holly|    white| 43|0.24|     null|
    //  |   luke|steinbeck| 51|1.14|    world|
    //  |   nick|underhill| 31|2.46|     null|
    //  |  holly| davidson| 59|1.26|    where|
    //  | calvin|    brown| 56|0.72|     null|
    //  | rachel| robinson| 62|2.25|     null|
    //  |    tom|   carson| 35|0.56|    hello|
    //  +-------+---------+---+----+---------+

    innerSource.na.fill("").show()
    //  +-------+---------+---+----+---------+
    //  |   name|    alias|age| pga|otherInfo|
    //  +-------+---------+---+----+---------+
    //  |ulysses| thompson| 64| 1.9|         |
    //  |  katie|   carson| 25|3.65|         |
    //  |   luke|     king| 65|0.73|         |
    //  |  holly| davidson| 57|2.43|    hello|
    //  |   fred|   miller| 55|3.77|         |
    //  |  holly|    white| 43|0.24|         |
    //  |   luke|steinbeck| 51|1.14|    world|
    //  |   nick|underhill| 31|2.46|         |
    //  |  holly| davidson| 59|1.26|    where|
    //  | calvin|    brown| 56|0.72|         |
    //  | rachel| robinson| 62|2.25|         |
    //  |    tom|   carson| 35|0.56|    hello|
    //  +-------+---------+---+----+---------+

    // 4.当某行中otherInfo列的值是null或者NaN的时候丢弃此行
    innerSource.na.drop("all", List("otherInfo")).show()
    //  +-----+---------+---+----+---------+
    //  | name|    alias|age| pga|otherInfo|
    //  +-----+---------+---+----+---------+
    //  |holly| davidson| 57|2.43|    hello|
    //  | luke|steinbeck| 51|1.14|    world|
    //  |holly| davidson| 59|1.26|    where|
    //  |  tom|   carson| 35|0.56|    hello|
    //  +-----+---------+---+----+---------+
  }


  @Test
  def nullAndNaN(): Unit = {

    // 2.导入数据集

    // 3.读取数据集
    //    3.1.通过spark-csv自动的推断类型来读取, 推断数字的时候, 会将含有NaN那一列推断出字符串
    //id,year,month,day,hour,season,pm
    //val dataFrame = spark.read
    //  .option("inferSchema", value = true) // 类型自动推断
    //  .option("header", value = true)
    //  .csv("dataset/beijingpm_with_nan.csv")

    //    3.2.直接读取字符串, 在后序的操作中使用map算子转换类型
    //  spark.read.csv().map(row => row...)

    //    3.3.指定Schema, 不要自动推断
    val structType = StructType(
      List(
        StructField("id", LongType),
        StructField("year", IntegerType),
        StructField("month", IntegerType),
        StructField("day", IntegerType),
        StructField("hour", IntegerType),
        StructField("season", IntegerType),
        StructField("pm", DoubleType)
      )
    )

    val sourceDF = spark.read
      .option("header", value = true)
      .schema(structType)
      .csv("dataset/beijingpm_with_nan.csv")

    sourceDF.printSchema()

    sourceDF.show(100)

    // 4.丢弃
    // 39024,2014,6,14,23,2,130.0
    // 39025,2014,6,14,23,2,NaN
    //    规则:
    //        1.any, 当某行数据任意一个字段为NaN或null就丢弃此行  ,不指定就是any
    sourceDF.na.drop("any").show()
    sourceDF.na.drop().show()
    //        2.all, 当某行数据所有值都是 null 或者 NaN 的时候丢弃此行
    sourceDF.na.drop("all").show()
    //        3.某些列的规则, 当某行中特定列所有值都是null或者NaN的时候丢弃此行
    sourceDF.na.drop("all", List("year", "month", "day", "hour", "season")).show()
    //        4.当某行中特定列任意一个字段是null或者NaN的时候丢弃此行
    sourceDF.na.drop("any", List("year", "month", "day", "hour", "season")).show()

    // 5.填充
    // 规则:
    //      1.针对所有列数据进行默认值填充
    sourceDF.na.fill(0).show()
    //      2.针对特定列填充
    sourceDF.na.fill(0, List("year", "month")).show()
  }


  @Test
  def strProcessor(): Unit = {
    // 读取数据集
    val sourceDF = spark.read
      .option("header", value = true)
      .option("inferSchema", value = true)
      .csv("dataset/BeijingPM20100101_20151231.csv")

    //sourceDF.show()

    // 1.丢弃
    println("strProcessor 丢弃数据")
    sourceDF.where($"PM_Dongsi" =!= "NA").show()

    // 2.转换
    //import org.apache.spark.sql.functions._
    import org.apache.spark.sql.functions._

    // select name, age, case
    // when ... then ...
    // when ... then ...
    // else
    println("strProcessor 替换数据")
    sourceDF.select(
      $"No".as("id"), $"year", $"month", $"day", $"hour", $"season",
      when($"PM_Dongsi" === "NA", Double.NaN)
        .otherwise($"PM_Dongsi" cast DoubleType)
        .as("pm")
    ).show()


    // 使用DataFrameNaFunctions替换, 原类型和准换过后的类型, 必须一致
    println("strProcessor 替换数据02")
    sourceDF.na
      .replace("PM_Dongsi", Map("NA" -> "NaN", "NULL" -> "null"))
      .show()
  }
}

